package cc.unmi;

import cc.unmi.serialization.AvroSerializer;
import org.apache.avro.specific.SpecificRecordBase;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * @note 使用传统的 avro API 自定义序列化类和反序列化类
 * @ref  https://www.jianshu.com/p/4f724c7c497d
 */
public class AvroProducer<T extends SpecificRecordBase> {

    private KafkaProducer<String, T> producer = new KafkaProducer<>(getProperties());

    public void sendData(Topic topic, T data) {
        producer.send(new ProducerRecord<>(topic.topicName, data),
                (metadata, exception) -> {
                    if (exception == null) {
                        System.out.printf("Sent user: %s, metadata: partition: %s, offset: %s\n", data, metadata.partition(), metadata.offset());
                    } else {
                        System.out.println("data sent failed: " + exception.getMessage());
                    }
                });
    }

    private Properties getProperties() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, Topic.SERVERS_CONFIG);
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "DemoProducer");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                AvroSerializer.class.getName());
        return props;
    }
}
